package com.autonavi.yunda.yunji.core.service.dbhelper;

import com.autonavi.yunda.yunji.common.thread.TraceableThreadPoolTaskExecutor;
import com.autonavi.yunda.yunji.core.utils.diling.ObjectUtil;
import com.google.common.collect.Lists;
import com.taobao.tair.DataEntry;
import com.taobao.tair.Result;
import com.taobao.tair.ResultCode;
import com.taobao.tair.impl.mc.MultiClusterTairManager;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Future;


@Slf4j
public class TairHelper {
    private static final int try_cnt = 2;
    private final Integer namespace;
    private final MultiClusterTairManager tairManager;
    private final TraceableThreadPoolTaskExecutor tairAsyncExecutor;

    public TairHelper(Integer namespace,
                      MultiClusterTairManager tairManager,
                      TraceableThreadPoolTaskExecutor tairAsyncExecutor) {
        this.namespace = namespace;
        this.tairManager = tairManager;
        this.tairAsyncExecutor = tairAsyncExecutor;
    }

    public void close() {
        tairManager.close();
    }

    public ResultCode put(String key, String value, int version, int expired) {
        tairManager.setCompressionThreshold(1024 * 16);
        return tairManager.put(namespace, key, value, version, expired);
    }

    public String get(String key) {
        try {
            for (int i = 1; i <= try_cnt; i++) {
                Result<DataEntry> result = tairManager.get(namespace, key);
                if (result.getRc().equals(ResultCode.SUCCESS)) {
                    return String.valueOf(result.getValue().getValue());
                }
                if (result.getRc().equals(ResultCode.DATANOTEXSITS)) {
                    return null;
                }
                log.info("tair get failed, trycnt={}, key={}, rc={}", i, key, result.getRc());
            }
            log.info("tair重试读取返回空,key={}", key);
            return null;
        } catch (Exception e) {
            log.error("tair读取失败,key={}", key, e);
        }
        return null;
    }

    public boolean invalid(String key) {
        try {
            log.info("invalid:" + key);
            ResultCode ret = tairManager.invalid(namespace, key);
            log.info("invalid-end:" + ret.toString());
            return true;
        } catch (Exception e) {
            log.error("invalid-exception:" + key + "," + e.getMessage(), e);
        }
        return false;
    }

    public List<DataEntry> mget(List<String> key) {
        try {
            for (int i = 1; i <= try_cnt; i++) {
                Result<List<DataEntry>> result = tairManager.mget(namespace, key);
                if (result.getRc().equals(ResultCode.SUCCESS) || result.getRc().equals(ResultCode.PARTSUCC)) {
                    return result.getValue();
                }
                log.info(key + " retcode " + result.getRc());
                if (result.getRc().equals(ResultCode.DATANOTEXSITS)) {
                    log.info(key + " DATA NOT EXSITS");
                    return null;
                }
            }
            log.info("tair重试读取返回空,key={}", key);
            return null;
        } catch (Exception e) {
            log.error("tair读取失败,key={}", key, e);
        }
        return null;
    }

    /**
     * 异常返回-1
     */
    public Map<String, Integer> mgetInt(List<String> key) {
        Map<String, Integer> rqpMap = new HashMap<>(8);
        if (key == null || key.size() == 0) {
            return rqpMap;
        }
        try {
            List<DataEntry> qpInfo = mget(key);
            if (qpInfo != null && qpInfo.size() > 0) {
                for (DataEntry dataEach : qpInfo) {
                    String keyName = dataEach.getKey().toString();
                    int val = -1;
                    Object valS1 = dataEach.getValue();
                    if (ObjectUtil.isNotEmpty(valS1)) {
                        try {
                            val = Integer.parseInt(valS1.toString());
                        } catch (Exception e) {
                            log.error("mgetInt exception:" + e.getMessage(), e);
                        }
                    }
                    rqpMap.put(keyName, val);
                }
            }
        } catch (Exception e) {
            log.error("tair-Getint Exception:" + e.getMessage(), e);
        }
        for (String k : key) {
            if (!rqpMap.containsKey(k)) {
                rqpMap.put(k, -1);
            }
        }
        return rqpMap;
    }

    public int incTime(String key, Integer expireTime) {
        Result<Integer> re = tairManager.incr(namespace, key, 1, 0, expireTime);
        return re.getValue();
    }

    public List<DataEntry> tairBatchGet(List<String> keys, int groupSize) {
        List<DataEntry> values = new ArrayList<>();
        List<Future<List<DataEntry>>> tasks = new ArrayList<>();
        List<List<String>> keysGroup = Lists.partition(keys, groupSize);
        long startTime = System.nanoTime();
        for (List<String> subKeys : keysGroup) {
            Future<List<DataEntry>> task = tairAsyncExecutor.submit(() -> mget(subKeys));
            tasks.add(task);
        }
        List<DataEntry> list;
        for (Future<List<DataEntry>> t : tasks) {
            try {
                list = t.get();
                if (CollectionUtils.isEmpty(list)) {
                    log.error("未查询到该key的value");
                } else {
                    values.addAll(t.get());
                }
            } catch (Exception e) {
                log.error("异步获取tair value异常:", e);
            }
        }
        long endTime = System.nanoTime();
        long duration = (endTime - startTime) / 1000000;
        log.info("batchGet[keySize:{},retSize:{},elapsed:{}ms,key:{}]", keys.size(), values.size(), duration, keys.get(0));
        return values;
    }
}
